package org.lqk.kafka.producer;




import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;
import java.util.Random;

/**
 * Created by bert on 16/10/18.
 */
public class DemoProducer {

    /*
        由此可以看到,关键的步骤是

        1. 生成ProducerConfig对象
        2. 生成producer对象
     */

    public static void main(String[] args) {
        String topic= "test";
        long events = 100;
        Random rand = new Random();

        Properties props = new Properties();
        props.put("metadata.broker.list", "localhost:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, String> producer = new Producer<String, String>(config);

        /*
            Kafka中可以将Topic从物理上划分成一个或多个分区（Partition），
            每个分区在物理上对应一个文件夹，以”topicName_partitionIndex”的命名方式命名，
            该文件夹下存储这个分区的所有消息(.log)和索引文件(.index)，这使得Kafka的吞吐率可以水平扩展。

            生产者在生产数据的时候，可以为每条消息指定Key，这样消息被发送到broker时，
            会根据分区规则选择被存储到哪一个分区中，如果分区规则设置的合理，
            那么所有的消息将会被均匀的分布到不同的分区中，这样就实现了负载均衡和水平扩展。
         */
        for (long nEvents = 0; nEvents < events; nEvents++) {
            String msg = "NativeMessage-" + rand.nextInt() ;
            KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, nEvents + "", msg);
            producer.send(data);
        }
        producer.close();

    }
}

